package vip.xiaomoli.courier.component;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import vip.xiaomoli.courier.component.timewheel.CourierTask;
import vip.xiaomoli.courier.component.timewheel.CourierTimeWheel;
import vip.xiaomoli.courier.dao.CourierDao;
import vip.xiaomoli.courier.model.TaskInfo;
import vip.xiaomoli.courier.util.ApplicationContextHelper;
import vip.xiaomoli.courier.util.CourierConstant;
import vip.xiaomoli.courier.util.CourierUtil;

import java.util.Date;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

public class CourierDBScanner {

    private static final Logger LOG = LoggerFactory.getLogger(CourierDBScanner.class);

    private volatile int pos = -1;

    private volatile int clusterNum = 0;

    private volatile List<String> clusterIps;

    private CourierDao dao = ApplicationContextHelper.getBean(CourierDao.class);

    private CourierTimeWheel timeWheel = ApplicationContextHelper.getBean(CourierTimeWheel.class);

    private ScheduledExecutorService dbScanner = Executors.newSingleThreadScheduledExecutor();

    private ScheduledExecutorService dbCleaner = Executors.newSingleThreadScheduledExecutor();

    public CourierDBScanner() {
        start();
    }

    private void start() {
        LOG.info("CourierScanner Starting...");
        dao.unloadTasks(CourierUtil.getLocalIP()); // 卸载本机之前未完成的任务，以免这台机器是集群中最后一个宕机的
        dbScanner.scheduleWithFixedDelay(
                () -> {
                    if (clusterNum > 0 && pos > -1) {
                        while (true) {
                            List<TaskInfo> list = dao.listTaskInfo4Exec(
                                    CourierUtil.addSeconds(new Date(), CourierConstant.ADVANCE_LOAD_SECONDS), // execTimeUB
                                    CourierUtil.addSeconds(new Date(), -CourierConstant.FAIL_RETRY_SECONDS), // execEndTimeUB
                                    CourierUtil.addSeconds(new Date(), -CourierConstant.EXEC_TIMEOUT_SECONDS), // loadTimeUB
                                    clusterNum, pos
                            );
                            if (0 == list.size()) break;
                            for (TaskInfo ti : list) {
                                ti.setExecServerIP(CourierUtil.getLocalIP());
                                dao.recordTaskLoad(ti);
                                int delaySeconds = CourierUtil.getSeconds(ti.getExecTime(), new Date());
                                timeWheel.addTask(new CourierTask(ti, dao), delaySeconds);
                            }
                        }
                    }
                },
                1, CourierConstant.SCAN_DELAY_SECONDS, TimeUnit.SECONDS
        );
        dbCleaner.scheduleWithFixedDelay(
                () -> {
                    while (true) {
                        int num = dao.deleteSucc(CourierUtil.addDays(new Date(), -CourierConstant.SUCC_RETAIN_DAYS));
                        if (num <= 0) break;
                    }
                },
                3600, 3600, TimeUnit.SECONDS
        );
        LOG.info("CourierScanner Started.");
    }

    public void setPos(int pos) {
        this.pos = pos;
    }

    public void setClusterIps(List<String> clusterIps) {
        for (String ip : this.clusterIps) {
            if (!clusterIps.contains(ip)) dao.unloadTasks(ip); // 卸载掉已下线的机器的任务
        }
        this.clusterIps = clusterIps;
        this.clusterNum = clusterIps.size();
    }

}
